package com.chengzhi.disruptor.quickstart;

import com.chengzhi.disruptor.quickstart.group.OrderGroupEventHandlerOne;
import com.chengzhi.disruptor.quickstart.group.OrderGroupEventHandlerThree;
import com.chengzhi.disruptor.quickstart.group.OrderGroupEventHandlerTwo;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

public class OrderEventTest {
    public static void main(String[] args) {
        OrderEventFactory orderEventFactory = new OrderEventFactory();
        /**
         * 消息工厂对象  容量   线程池  生产者(单利/多利)   等待策略
         */
        Disruptor<OrderEvent> disruptor = new Disruptor<>(orderEventFactory, 1024 * 1024, r -> {
            return new Thread(r);
        }, ProducerType.SINGLE, new BlockingWaitStrategy());
        //广播单消费者监听
//        disruptor.handleEventsWith(new OrderEventHandler());
        //分组消费者
        disruptor.handleEventsWithWorkerPool(new OrderGroupEventHandlerOne(), new OrderGroupEventHandlerTwo(), new OrderGroupEventHandlerThree());
        disruptor.start();

        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
        OrderEventProducer producer = new OrderEventProducer(ringBuffer);
        ByteBuffer buffer = ByteBuffer.allocate(8);
        for (long i = 0; i < 1000; i++) {
            buffer.putLong(0, i);
            producer.send(buffer);
        }
        disruptor.shutdown();
    }
}
